package com.fangsheng.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.jupiter.api.Test;

import java.util.List;

/**
 * @创建人: 放生
 * @创建时间: 2022/1/22
 * @描述: 自定义选择器 优化于 ExampleRocketMQ 类中的案列 2
 */
public class ConsumerSelectOrderQueue {


    //--------------example 002------------------------

    /**
     * 保证有序性：
     * 1，producer，其实就是自定义分区器(队列选择器)
     * 2，consumer，MessageListenerOrderly，
     * 感知：
     * 才能实现，生产的时候，有顺序依赖的msg进入一个队列，并且，消费者也会再多线程情况下保证单线程对应的queue能按顺序消费
     * 如果，你只创建一个queue，全局有序
     * @throws Exception
     */
    @Test
    public void orderProducer() throws Exception {

        DefaultMQProducer producer = new DefaultMQProducer("ooxx");
        producer.setNamesrvAddr("192.168.150.11:9876;192.168.150.12:9876");
        producer.start();
        // 尽可能把 new 对象放在循坏外面
        MyMessageQueueSelector selector = new MyMessageQueueSelector();
        for (int i = 0; i < 10; i++) {

            Message msg = new Message(
                    "order_topic",
                    "TagA",
                    ("message body : " + i + "type: " + i % 3).getBytes()
            );

            //TODO
            SendResult res = producer.send(msg, selector, i % 3);
            System.out.println(res);
        }


    }

    @Test
    public void orderConsumer() throws Exception {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("xoxo03");
        consumer.setNamesrvAddr("192.168.10.11:9876;192.168.10.21:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        try {
            consumer.subscribe("order_topic","*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {

                msgs.forEach(msge->{
                    System.out.println(Thread.currentThread().getName()+ " : "+new String(msge.getBody()));
                });
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();
        System.in.read();
        consumer.shutdown();
    }





}
